package bleudp

import (
	"context"
	"encoding/json"
	"fmt"
	"iot-ble-server/global/globalconstants"
	"iot-ble-server/global/globallogger"
	"iot-ble-server/global/globalmemo"
	"iot-ble-server/global/globalredis"
	"iot-ble-server/global/globalstruct"
	"iot-ble-server/global/globalutils"
	"iot-ble-server/internal/config"
	"iot-ble-server/internal/packets"
	"math/rand"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/redis/go-redis/v9"
)

//创建全局mqtt publish消息处理 handler
var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
	globallogger.Log.Warnf("Sub Client Unknow Topic: %s, msg : %s", msg.Topic(), msg.Payload())
}

var ClientOptions *mqtt.ClientOptions
var client mqtt.Client
var taskID = "h3c-iot-ble-server"
var connectting = false
var connectFlag = false

func connect(clientOptions *mqtt.ClientOptions) error {
	//创建客户端连接
	client = mqtt.NewClient(clientOptions)
	//客户端连接判断
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		globallogger.Log.Errorf("[MQTT] mqtt connect error, taskID: %s, error: %s", taskID, token.Error())
		return token.Error()
	} else {
		globallogger.Log.Errorf("[MQTT] connect success taskID:%s, token:%+v, clientOptions:%+v", taskID, token, clientOptions)
		return nil
	}
}
func connectLoop(clientOptions *mqtt.ClientOptions) {
	defer func() {
		err := recover()
		if err != nil {
			globallogger.Log.Errorln("MQTT connectLoop err :", err)
		}
	}()
	for {
		if err := connect(clientOptions); err != nil {
			client.Disconnect(250)
			time.Sleep(time.Second * 2)
		} else {
			connectFlag = true
			connectting = false
			break
		}
	}
}

// ConnectMQTT ConnectMQTT
func ConnectMQTT(mqttHost string, mqttPort string, mqttUserName string, mqttPassword string) {
	defer func() {
		err := recover()
		if err != nil {
			globallogger.Log.Errorln("ConnectMQTT err :", err)
		}
	}()
	//设置连接参数
	ClientOptions = mqtt.NewClientOptions().AddBroker("tcp://" + mqttHost + ":" + mqttPort).SetUsername(mqttUserName).SetPassword(mqttPassword)
	//设置客户端ID
	ClientOptions.SetClientID(fmt.Sprintf("%s-%d", taskID, rand.Int()))
	//设置handler
	ClientOptions.SetDefaultPublishHandler(messagePubHandler)
	//设置保活时长
	ClientOptions.SetKeepAlive(30 * time.Second)
	ClientOptions.SetAutoReconnect(true)
	ClientOptions.SetMaxReconnectInterval(time.Minute)
	connectting = true
	connectLoop(ClientOptions)
	go func() {
		for {
			if !connectting && !client.IsConnectionOpen() {
				connectting = true
				client.Disconnect(250)
				connectLoop(ClientOptions)
			}
			time.Sleep(2 * time.Second)
		}
	}()
}

// Publish Publish
func Publish(topic string, msg string) {
	if client.IsConnectionOpen() {
		//发布消息
		token := client.Publish(topic, 0, false, msg)
		if !token.WaitTimeout(60 * time.Second) {
			globallogger.Log.Errorf("[MQTT] Publish timeout: topic: %s, mqttMsg %s", topic, msg)
			client.Disconnect(250)
			time.Sleep(time.Second)
			return
		}
		if token.Error() != nil {
			globallogger.Log.Errorf("[MQTT] Publish error: topic: %s, mqttMsg %s, error %s", topic, msg, token.Error().Error())
			client.Disconnect(250)
			time.Sleep(time.Second)
		} else {
			globallogger.Log.Warnf("[MQTT] Publish success: topic: %s, mqttMsg %s", topic, msg)
		}
	} else {
		globallogger.Log.Errorf("[MQTT] Publish client is not connect: topic: %s, mqttMsg %s", topic, msg)
	}
}

// Subscribe Subscribe
func Subscribe(topic string, cb func(topic string, msg []byte)) {
	globallogger.Log.Errorf("Sub Topic : %s start", topic)
	client.Subscribe(topic, 1, func(client mqtt.Client, msg mqtt.Message) {
		globallogger.Log.Warnf("Sub Client Topic : %s, msg : %s", msg.Topic(), msg.Payload())
		go cb(msg.Topic(), msg.Payload())
	})
}

func GetConnectFlag() bool {
	return connectFlag
}

func SetConnectFlag(flag bool) {
	connectFlag = flag
}

func StartMQ(ctx context.Context) error {
	globallogger.Log.Errorln("Start ble mqtt server")
	go ConnectMQTT(config.C.MQTTConfig.MqttHost, config.C.MQTTConfig.MqttPort, config.C.MQTTConfig.Username, config.C.MQTTConfig.Password)
	go func() {
		for {
			if GetConnectFlag() {
				globallogger.Log.Warnln("start mqtt subscribe")
				subscribeFromMQTT()
				SetConnectFlag(false)
			}
			time.Sleep(10 * time.Second)
		}
	}()
	return nil
}

func subscribeFromMQTT() {
	subTopics := make([]string, 0)
	subTopics = append(subTopics, globalconstants.TopicV3GatewayRPC)
	subTopics = append(subTopics, globalconstants.TopicV3DeviceRPC)
	subTopics = append(subTopics, globalconstants.TopicV3GatewayEventDeviceDelete)
	for _, topic := range subTopics {
		Subscribe(topic, func(topic string, msg []byte) {
			ProcSubMsg(topic, msg)
		})
	}
}

func ProcSubMsg(topic string, jsonMsg []byte) {
	defer func() {
		err := recover()
		if err != nil {
			globallogger.Log.Errorln("ProcSubMsg err :", err)
		}
	}()
	switch topic {
	case globalconstants.TopicV3GatewayRPC:
		var mqttMsg globalstruct.RPCGatewayIotware
		err := json.Unmarshal(jsonMsg, &mqttMsg)
		if err != nil {
			globallogger.Log.Errorln("[Subscribe]: JSON gateway rpc marshaling failed:", err)
			return
		}
		globallogger.Log.Warnf("[Subscribe]: mqtt msg subscribe: topic: %s, mqttMsg %+v", topic, mqttMsg)
		procGatewayRPCIotware(mqttMsg)
	case globalconstants.TopicV3DeviceRPC:
		var mqttMsg globalstruct.RPCIotware
		err := json.Unmarshal(jsonMsg, &mqttMsg)
		if err != nil {
			globallogger.Log.Errorln("[Subscribe]: JSON rpc marshaling failed:", err)
			return
		}
		globallogger.Log.Warnf("[Subscribe]: mqtt msg subscribe: topic: %s, mqttMsg %+v", topic, mqttMsg)
		procRPCIotware(mqttMsg)
	case globalconstants.TopicV3GatewayEventDeviceDelete:
		var deleteMqttMsg globalstruct.EventDeviceDeleteIotware
		err := json.Unmarshal(jsonMsg, &deleteMqttMsg)
		if err != nil {
			globallogger.Log.Errorln("[Subscribe]: JSON delete marshaling failed:", err)
			return
		}
		globallogger.Log.Warnf("[Subscribe]: mqtt msg subscribe: topic: %s, mqttMsg %+v", topic, deleteMqttMsg)
		for _, devMac := range deleteMqttMsg.DeviceList {
			procTerminalDelete(devMac)
		}
	default:
		globallogger.Log.Warnf("[Subscribe]: invalid topic: %s", topic)
	}

}

//网关模块复位请求
func procGatewayRPCIotware(mqttMsg globalstruct.RPCGatewayIotware) {
	ctx := context.Background()
	moduleId := globalutils.ConvertDecimalToHexStr(mqttMsg.Data.Para.IotModuleId, globalconstants.BYTE_STR_FOUR)
	jsonInfo, err := GenerateDownJsonInfo(ctx, &globalstruct.DownJsonInfoUnify{
		GwMac:         mqttMsg.Device,
		IotModuleId:   moduleId,
		LinkedMsgType: packets.IotModuleRset,
		Ctrl:          globalconstants.CtrlLinkedMsgHeadWithBoy,
	})
	if err != nil {
		globallogger.Log.Errorf("<procGatewayRPCIotware> [DevEui %s] generate module reset info failed", mqttMsg.Device)
		return
	}
	byteToSend, err1 := EnCodeForDownUdpMessage(jsonInfo)
	if err1 != nil {
		globallogger.Log.Errorf("<procGatewayRPCIotware> [DevEui %s] marshal info failed", mqttMsg.Device)
		return
	}
	SendMsgBeforeDown(ctx, byteToSend, jsonInfo.MessageHeader.LinkMsgFrameSN, "", "", mqttMsg.Device, moduleId, packets.IotModuleRset)
}

//终端设备删除
//特征值以及服务需要删除
func procTerminalDelete(devMac string) {
	ctx := context.Background()
	var err error
	connCacheKey := globalutils.CreateCacheKey(globalconstants.BleDevCacheConnPrefix, devMac)
	msgCacheKey := globalutils.CreateCacheKey(globalconstants.BleDevCacheMessagePrefix, devMac)
	serviceCacheKey := globalutils.CreateCacheKey(globalconstants.DEV_SERVICE, devMac) //主服务对应的部分
	characterCacheKey := globalutils.CreateCacheKey(globalconstants.BleDevSeviceToCharacterPrifix, devMac)
	serviceUUIDCacheKey := globalutils.CreateCacheKey(globalconstants.BleDevCharacterToSevicePrifix, devMac)
	globalmemo.MemoCacheDownPriority.Remove(devMac)
	if config.C.General.UseRedis {
		_, err = globalredis.RedisCache.Del(ctx, connCacheKey, msgCacheKey, serviceCacheKey, characterCacheKey, serviceUUIDCacheKey).Result()
		if err != nil {
			globallogger.Log.Errorln("[procTerminalDelete]: JSON delete key failed:", err)
		}
	} else {
		globalmemo.BleFreeCacheDevConnInfo.Del([]byte(connCacheKey))
		globalmemo.MemoCacheDev.Remove(msgCacheKey)
		globalmemo.MemoCacheService.Remove(serviceCacheKey)
		globalmemo.BleCacheServiceForChar.Del([]byte(serviceUUIDCacheKey))
		globalmemo.BleCacheCharacterForSvc.Del([]byte(characterCacheKey))
	}
}

//特征操作请求
func procRPCIotware(mqttMsg globalstruct.RPCIotware) {
	ctx := context.Background()
	connCacheKey := globalutils.CreateCacheKey(globalconstants.BleDevCacheConnPrefix, mqttMsg.Device)
	var (
		terminal     globalstruct.TerminalInfo
		err          error
		strTerminal  string
		byteTerminal []byte
	)
	if config.C.General.UseRedis {
		strTerminal, err = globalredis.RedisCache.Get(ctx, connCacheKey).Result()
		if err != nil {
			if err != redis.Nil {
				globallogger.Log.Errorf("<procRPCIotware> DevEui %s redis has error %v\n", mqttMsg.Device, err)
				return
			}
			goto NoCache
		}
		err = json.Unmarshal([]byte(strTerminal), &terminal)
	} else {
		byteTerminal, err = globalmemo.BleFreeCacheDevConnInfo.Get([]byte(connCacheKey))
		if err != nil {
			goto NoCache
		}
		err = json.Unmarshal(byteTerminal, &terminal)
	}
	if err != nil {
		globallogger.Log.Errorf("<procRPCIotware> DevEui %s unmarshal failed %v", mqttMsg.Device, err)
		return
	}
NoCache:
	priorityNode := globalstruct.TerminalInfo{
		RSSI: -128,
	}
	if terminal.ConnectStatus == "" || terminal.ConnectStatus == "00" { //连接不存在或者连接状态为断开
		gwInfo, exist := globalmemo.MemoCacheDownPriority.Get(mqttMsg.Device)
		if !exist {
			globallogger.Log.Errorf("<procRPCIotware>[DevEui %s] can't find dev gateway infomation", mqttMsg.Device)
			return
		}
		for _, node := range gwInfo.(map[string]globalstruct.TerminalInfo) {
			if node.RSSI > priorityNode.RSSI {
				priorityNode = node
			}
		}
		err = ConnectBleDev(ctx, priorityNode)
		if err != nil {
			globallogger.Log.Errorf("<procRPCIotware> [DevEui %s] connect failed %v", mqttMsg.Device, err)
			return
		}
	}
	err = CharacterOptBleDev(ctx, terminal, mqttMsg.Data)
	if err != nil {
		globallogger.Log.Errorf("<procRPCIotware> [DevEui :%s] has error %v", mqttMsg.Device, err)
	}
}
